package main;

import helper.CompositeKey;
import helper.CompositeKeyComparator;
import helper.PairGroupingComparator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class Main {

    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        JobConf jobConf = new JobConf(new Configuration(), Main.class);
        Job job = Job.getInstance(jobConf, "SimilarityCount");
        job.setJarByClass(Main.class);
        
        jobConf.setInt("mapreduce.tasktracker.map.tasks.maximum", 32);
        jobConf.setInt("mapreduce.tasktracker.reduce.tasks.maximum", 32);
        job.setNumReduceTasks(Integer.parseInt(args[2]));
        FileInputFormat.setMaxInputSplitSize(job, Integer.parseInt(args[3]));
//        jobConf.setNumReduceTasks(4);
//        jobConf.set("mapred.min.split.size", args[3]);
        //Define the comparator that controls which keys are grouped together 
        //for a single call to combiner, Reducer.reduce
        job.setCombinerKeyGroupingComparatorClass(PairGroupingComparator.class);
        
        //Define the comparator that controls which keys are grouped together 
        //for a single call to Reducer.reduce
        job.setGroupingComparatorClass(PairGroupingComparator.class);
        
        //Define the comparator that controls how the keys are sorted 
        //before they are passed to the Reducer.
        job.setSortComparatorClass(CompositeKeyComparator.class);
        
        job.setMapperClass(PairMapper.class);
        job.setPartitionerClass(PairPartitioner.class);
        job.setCombinerClass(ValueCombiner.class);
        job.setReducerClass(SortReducer.class);
        
        job.setMapOutputKeyClass(CompositeKey.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);
        
    }

}
